package com.ziroom.zifamily.zfcensor.common.config.rocketmq;

import com.alibaba.rocketmq.client.consumer.DefaultMQPushConsumer;
import com.alibaba.rocketmq.client.exception.MQClientException;
import com.alibaba.rocketmq.common.consumer.ConsumeFromWhere;
import com.ziroom.zifamily.zfcensor.common.exception.RocketMQException;
import com.ziroom.zifamily.zfcensor.mq.RocketConsumerListener;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.SpringBootConfiguration;
import org.springframework.context.annotation.Bean;

/**
 * https://www.jianshu.com/p/8c4c2c2ab62e
 * <p>
 * <PRE>
 * <BR>    修改记录
 * <BR>-----------------------------------------------
 * <BR>    修改日期         修改人          修改内容
 * </PRE>
 *
 * @author cuiyh9
 * @version 1.0
 * @Date Created in 2018年11月29日 16:48
 * @since 1.0
 */
@Slf4j
@SpringBootConfiguration
public class RocketMQConsumerConfiguration {


    @Value("${rocketmq.consumer.groupName}")
    private String groupName;

    @Value("${rocketmq.consumer.namesrvAddr}")
    private String namesrvAddr;

    @Value("${rocketmq.consumer.topics}")
    private String topics;

    @Autowired
    private RocketConsumerListener rocketConsumerListener;

    @Bean(initMethod = "start", destroyMethod = "shutdown")
    public DefaultMQPushConsumer defaultMQPushConsumer() {

        if (StringUtils.isEmpty(groupName)
                || StringUtils.isEmpty(namesrvAddr)
                || StringUtils.isEmpty(topics)) {
            throw new RocketMQException("MQ消费者groupName或namesrvAddr或topics未配置");
        }
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(groupName);
        consumer.setNamesrvAddr(namesrvAddr);
        try {
            String[] topicsArry = topics.split(";");
            for (String topic : topicsArry) {
                consumer.subscribe(topic, "*");
            }

            consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
            consumer.registerMessageListener(rocketConsumerListener);
        } catch (MQClientException e) {
            log.error("create defaultMQPushConsumer error", e);
        }
        return consumer;
    }

}
